package com.huanwang.commonlibrary.utils.rxjava;

import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.processors.FlowableProcessor;
import io.reactivex.rxjava3.processors.PublishProcessor;
import io.reactivex.rxjava3.schedulers.Schedulers;


/**
 * @author wang_hong_wei
 * @description Rajava 发送消息
 * @Date 2018-5-11
 */

public class RxBus {

    private static volatile RxBus sInstance;

    public static RxBus getInstance() {
        if (sInstance == null) {
            synchronized (RxBus.class) {
                if (sInstance == null) {
                    sInstance = new RxBus();
                }
            }
        }
        return sInstance;
    }


    private FlowableProcessor<Object> mEventBus = PublishProcessor.create().toSerialized();

    public void post(Object event) {
        mEventBus.onNext(event);
    }

    private  <T> Flowable<T> toObservable(Class<T> clz) {
        return mEventBus.ofType(clz).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
    }

    public Disposable getDisposable(Consumer consumer){
        return toObservable(Object.class).subscribe(consumer);
    }

    //---------------使用方法如下   Activity 中直接重写此方法 这是接收的消息
    /* public Consumer addRXBusListen(){
      return new Consumer() {
            @Override
            public void accept(Object o) throws Exception {
                if (o instanceof MyBean){
                    ToastUtils.bugToast("-zou-");
                }
            }
        };
    }*/

     //----------------发送的使用如下
    // RxBus.getInstance().post(new MyBean());
}
